package net.bw.realtime.jtp.common.utils;

import java.io.IOException;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.HashMap;
import java.util.Map;

/*
 * @ Author：liuyawei
 * @ Date：2025-06-04
 */
public class HBaseUtil {

        // 定义连接
        private static Connection connection = null;

        static {
                try{
                        // 1.创建配置对象，设置属性值，比如连接ZK集群信息
                        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create() ;
                        conf.set(HConstants.CLIENT_ZOOKEEPER_QUORUM, "node101,node102,node103");
                        conf.set(HConstants.CLIENT_ZOOKEEPER_CLIENT_PORT, "2181");
                        // 2.实例化Connection连接
                        connection = ConnectionFactory.createConnection(conf) ;
                }catch (Exception e){
                        e.printStackTrace();
                }
        }

        // 获取HBase表中的数据
        public static String getData(String tableName, String rowKey, String family, String column) throws IOException {

                // 获取table
                Table table = connection.getTable(TableName.valueOf(tableName));

                // 获取数据
                Get get = new Get(Bytes.toBytes(rowKey));
                Result result = table.get(get);

                // 获取字段值
                byte[] resultValue = result.getValue(Bytes.toBytes(family), Bytes.toBytes(column));

                // 关闭table
                table.close();

                // 返回字段值
                return Bytes.toString(resultValue);
        }

        // 扫码HBase表
        public static Map<String, String> scanData(String tableName, String family, String column) throws Exception{
                // a.实例化Table对象
                Table table = connection.getTable(TableName.valueOf(tableName));
                // b.构建扫描器对象
                Scan scan = new Scan() ;
                scan.addFamily(Bytes.toBytes(family)) ;
                // c. 扫描数据
                ResultScanner scanner = table.getScanner(scan);
                // d. 提取数据
                Map<String, String> dimMap = new HashMap<>() ;
                for (Result result : scanner) {
                        // RowKey行健
                        String rowKey = Bytes.toString(result.getRow());
                        // 列值
                        String columnValue = Bytes.toString(result.getValue(Bytes.toBytes(family), Bytes.toBytes(column)));
                        // 放入集合
                        dimMap.put(rowKey, columnValue);
                }
                // e. 关闭
                scanner.close();
                table.close();
                // f. 返回对象
                return dimMap ;
        }

        // 关闭连接
        public static void close() throws IOException {
                if(null != connection){
                        connection.close();
                }
        }

}
